package dongshi.daddy.juc;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.*;

/**
 * CompetableFuture测试类
 */
public class CompletableFutureTest {

    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>());

    public static void main(String[] args) throws Exception {
        defaultThread();
//        myThread();
//        runAsyncNoReturn();
//        runAsyncWithReturn();
//        thenApplyAsync();
//        thenAcceptAsync();
//        thenRunAsync();
//        thenCompose();
//        thenCombine();
//        thenAcceptBothAsync();
//        runAfterBothAsync();
//        applyToEither();
//        acceptEither();
//        runAfterEither();
//        allOf();
//        anyOf();
//        handle();
//        whenComplete();
//        exceptionally();
    }

    // 捕获任务异常并获取返回值
    public static void exceptionally() throws ExecutionException, InterruptedException {
        System.out.println("main.................start.....");
        // try catch的一种更优雅的写法
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 0;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).exceptionally(throwable -> {
            System.out.println("当前线程：" + Thread.currentThread().getId());
            //exceptionally可以感知错误并返回指定值
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {}
            System.out.println("执行了exceptionally");
            return 30;
        });
        System.out.println("main.................end....." + completableFuture.get());
    }


    public static void whenComplete() {
        System.out.println("main.................start.....");
        final CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 0;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).whenComplete((result, throwable) -> {
            //whenComplete虽然得到异常信息，但是不能修改返回信息
            System.out.println("异步完成。。。。结果是：" + result + "...异常是：" + throwable);
        });

        try {
            System.out.println("main.................end..T..." + completableFuture.get());
        } catch (InterruptedException e) {
            System.out.println("报错了1");
        } catch (ExecutionException e) {
            System.out.println("报错了2");
        }
    }


    // 感知异常：获取上一个任务的结果或者是异常
    public static void handle() throws ExecutionException, InterruptedException {
//        System.out.println("main.................start.....");
        final CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
//            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 0;
//            System.out.println("运行结果：" + i);
            return i;
        }, executor).handleAsync((in, throwable) -> {
            if (throwable != null) {
                System.out.println("上个任务异常了：" + throwable.getMessage());
                return "报错返回";
            }
            return "正确了";
        });
        System.out.println("main.................end....." + completableFuture.get());
    }


    // 多个任务任务，有个任务完成，则获取其结果
    public static void anyOf() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(new Random().nextInt(5));
            } catch (InterruptedException e) { }
            System.out.println("任务1完成");
            return "任务1";
        }, executor);
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(new Random().nextInt(5));
                System.out.println("任务2完成");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "任务2";
        }, executor);

        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(new Random().nextInt(5));
            } catch (InterruptedException e) { }
            System.out.println("任务3完成");
            return "任务3";
        }, executor);
        CompletableFuture<Object> anyOf = CompletableFuture.anyOf(future1, future2, future3);
        System.out.println("anyOf--最先完成的是" + anyOf.get());
        //等待future2打印
//        System.out.println("等等任务2");
//        Thread.sleep(3000);
    }


    // 所有任务执行完毕后，获取所有任务的执行结果
    public static void allOf() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务1完成");
            return "任务1";
        }, executor);
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
                System.out.println("任务2完成");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "任务2";
        }, executor);
        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务3完成");
            return "任务3";
        }, executor);

        CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2, future3);
        //等待所有任务完成
        allOf.join();
        System.out.println("所有任务完成，获取所有任务返回值" + "allOf" + future1.get() + "-------" + future2.get() + "-------" + future3.get());
    }

    // 有一个任务执行完成就执行，不获取前面任务返回值，最终不返回值
    public static void runAfterEither() {
        CompletableFuture<Object> future1 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("任务1线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            try {
                Thread.sleep(new Random().nextInt(5));
                System.out.println("任务1执行完成");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return i;
        }, executor);

        CompletableFuture<Object> future2 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("任务2线程：" + Thread.currentThread().getId());
            try {
                Thread.sleep(new Random().nextInt(5));
                System.out.println("任务2执行完成");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        }, executor);

        CompletableFuture<Void> runAfterEitherAsync = future1.runAfterEitherAsync(future2, () -> {
            System.out.println("任务3开始执行。。。");
        }, executor);
    }

    // 两个任务中其中一个执行完，第三个任务获取完成任务返回值执行，最终无返回值
    public static void acceptEither() {
        CompletableFuture<Object> future1 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("任务1线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            try {
                Thread.sleep(new Random().nextInt(5));
                System.out.println("任务1运行完成");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return i;
        }, executor);

        CompletableFuture<Object> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(new Random().nextInt(5));
                System.out.println("任务2运行完成");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        }, executor);

        CompletableFuture<Void> acceptEitherAsync = future1.acceptEitherAsync(future2, result -> {
            System.out.println("任务3开始执行。。。获取到上一个任务结果：" + result);
        }, executor);

    }

    // 两个任务中其中一个执行完，第三个任务获取完成任务返回值执行，最终有返回值
    public static void applyToEither() throws ExecutionException, InterruptedException {
        CompletableFuture<Object> future1 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("任务1线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            try {
                Thread.sleep(new Random().nextInt(5));
                System.out.println("任务1运行结束");
            } catch (InterruptedException e) {}
            return i;
        }, executor);

        CompletableFuture<Object> future2 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("任务2线程：" + Thread.currentThread().getId());
            try {
                Thread.sleep(new Random().nextInt(5));
            } catch (InterruptedException e) {}
            System.out.println("任务2执行结束");
            return "hello";
        }, executor);

        CompletableFuture<String> applyToEitherAsync = future1.applyToEitherAsync(future2, result -> {
            System.out.println("任务3开始执行。。。获取结果：" + result);
            return result.toString() + " world";
        }, executor);
        System.out.println("任务3结果：" + applyToEitherAsync.get());
    }

    // 两个任务完成后执行第三个任务
    public static void runAfterBothAsync() {
        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("任务1线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {}
            System.out.println("任务1执行结束");
            return i;
        }, executor);

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("任务2线程：" + Thread.currentThread().getId());
            System.out.println("任务2运行结束");
            return "hello";
        }, executor);

        CompletableFuture<Void> runAfterBothAsync = future1.runAfterBothAsync(future2, () -> {
            System.out.println("任务3启动。。。");
        }, executor);

    }

    // 结合两个线程的返回值，最终无返回值
    public static void thenAcceptBothAsync() throws Exception {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());

        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("任务1线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("任务1运行结束");
            return i;
        }, executor);

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("任务2线程：" + Thread.currentThread().getId());
            System.out.println("任务2运行结束");
            return "hello";
        }, executor);

        CompletableFuture<Void> thenAcceptBothAsync = future1.thenAcceptBothAsync(future2, (result1, result2) -> {
            System.out.println("任务3启动。。。结果1：" + result1 + "。。。结果2：" + result2);
        }, executor);

    }

    // 结合两个线程的返回值，最终有返回值
    public static void thenCombine() throws ExecutionException, InterruptedException {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());

        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("任务1线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("任务1运行结束");
            return i;
        }, executor);

        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
//            System.out.println("任务2线程：" + Thread.currentThread().getId());
            System.out.println("任务2运行结束");
            return "hello";
        }, executor);
        CompletableFuture<String> thenCombineAsync = future1.thenCombineAsync(future2, (result1, result2) -> {
            System.out.println("任务3启动。。。任务1结果：" + result1 + "。。。任务2结果：" + result2);
            return result2 + "-->" + result1;
        }, executor);
        System.out.println("任务3结果:" + thenCombineAsync.get());
    }

    public static void thenCompose() {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());

        CompletableFuture cf = CompletableFuture.completedFuture("hello")
                .thenCompose(str -> CompletableFuture.supplyAsync(() -> {
                    return str + ": thenCompose";
                }, executor));
        System.out.println(cf.join());
    }

    // 上一个任务执行完执行
    public static void thenRunAsync() throws Exception {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());

        System.out.println("main.................start.....");
        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).thenRunAsync(() -> {
            System.out.println("任务2启动了。。。。。");
        }, executor);
    }

    // 后一个线程获取前一个线程的执行的结果，没有返回值
    public static void thenAcceptAsync() throws Exception {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());

        CompletableFuture<Void> thenAcceptAsync = CompletableFuture.supplyAsync(() -> {
            System.out.println("thenAcceptAsync当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("thenAcceptAsync运行结果：" + i);
            return i;
        }, executor).thenAcceptAsync(result -> {
            System.out.println("thenAcceptAsync任务2启动了。。。。。上步结果：" + result);
        }, executor);
    }


    // 后一个线程获取前一个线程的执行的结果，最终有返回值
    public static void thenApplyAsync() throws Exception {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());

        CompletableFuture<String> thenApplyAsync = CompletableFuture.supplyAsync(() -> {
            System.out.println("thenApplyAsync当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("thenApplyAsync运行结果：" + i);
            return i;
        }, executor).thenApplyAsync(result -> {
            System.out.println("thenApplyAsync任务2启动了。。。。。上步结果：" + result);
            return "hello" + result * 2;
        }, executor);
        System.out.println("main.................end....." + thenApplyAsync.get());
    }


    // runAsync有返回值
    private static void runAsyncWithReturn() throws Exception {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());

                System.out.println("main.................start.....");
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getId());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
            }
            return i;
        }, executor);
        // 这里的get会阻塞等待
        System.out.println("main.................end....." + completableFuture.get());
    }


    // runAsync没有返回值
    public static void runAsyncNoReturn() {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
        System.out.println("main.................start.....");
        CompletableFuture.runAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
        }, executor);
        System.out.println("main.................end......");
        try {
            TimeUnit.SECONDS.sleep(5);
            System.exit(0);
        } catch (Exception e) {}
    }

    // 使用自定义的线程池执行任务
    public static void myThread() {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 10,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
        System.out.println("main.................start.....");
        CompletableFuture.runAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
        }, executor);
        System.out.println("main.................end......");
        try {
            TimeUnit.SECONDS.sleep(5);
            System.exit(0);
        } catch (Exception e) {}
    }


    // 使用默认的线程池commonPool执行任务
    public static void defaultThread() {
        System.out.println("main.................start.....");
        CompletableFuture.runAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
        });
        System.out.println("main.................end......");
        try {
            System.in.read();
        } catch (IOException e) {}
    }
}
